package com.chief.sink.elasticsearch;


import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.CreateRequest;
import co.elastic.clients.elasticsearch.core.CreateResponse;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;

public class EsSink extends RichSinkFunction<JSONObject> {

    ElasticsearchClient client = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        RestClient restClient = RestClient.builder(
                new HttpHost("hadoop52", 9200),
                new HttpHost("hadoop53", 9200),
                new HttpHost("hadoop54", 9200),
                new HttpHost("hadoop55", 9200)
        ).build();
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());
        client = new ElasticsearchClient(transport);
        BooleanResponse exists = client.indices().exists(e -> e.index("aaaaaaaaaaaaa"));
        if(!exists.value()) {
            CreateIndexResponse aaaaaaaaaaaaa = client.indices().create(e -> e.index("aaaaaaaaaaaaa"));
            System.out.println(aaaaaaaaaaaaa.acknowledged());
        }
    }

    @Override
    public void close() throws Exception {
        client.shutdown();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        System.out.println("sink47:"+value.toJSONString());
        CreateResponse createResponse = client.create(e -> e.index("aaaaaaaaaaaaa").id(String.valueOf(50 + (int) (Math.random() * 500)))
                .document(value));
        System.out.println("sink result is :"+createResponse.result());
    }
}
